package dht.simulation;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Vector;

import dht.data.Access;
import dht.data.AccessType;
import dht.data.Message;
import dht.data.MessageType;
import dht.statistics.DhtStatistics;

public class MasterNode {

	public int masterId;
	public double idleTime;
	public double splitTime;
	public RouteTableV2 routeTable;
	public HashSet<Integer> splitSourceList;
	public HashSet<Integer> splitDestinList;
	public static final double diskTime = GlobalVariables.masterDiskTime;

	public MasterNode(int masterId, int serverNum) {
		this.masterId = masterId;
		this.idleTime = 0.0;
		this.splitTime = 0.0;
		this.routeTable = new RouteTableV2(serverNum);
		this.splitSourceList = new HashSet<Integer>();
		this.splitDestinList = new HashSet<Integer>();
	}

	public Message masterStartAccess(Access access) throws DhtException {
		DhtStatistics.serverCpuReport(0, Math.max(idleTime, access.startTime)
				- idleTime, diskTime, Math.max(idleTime, access.startTime)
				+ diskTime);

		Message msg = new Message();
		msg.accessId = access.accessId;
		msg.fromId = access.requesterId;
		msg.generateTime = access.startTime;
		if (access.accessType == AccessType.VNODE_SPLIT) {
			msg.msgType = MessageType.SPLIT_REQ;
		}

		idleTime = Math.max(idleTime, msg.generateTime);
		msg.waitingCost = idleTime - msg.generateTime;
		msg.processCost = diskTime;
		msg.sendTime = msg.generateTime + msg.waitingCost + msg.processCost;
		idleTime = msg.sendTime;

		switch (msg.msgType) {
		case SPLIT_REQ:
			HashMap<Integer, Double> serverId2Load = routeTable
					.getServerLoads();
			HashMap<Integer, Integer> serverId2VnodeNum = routeTable
					.getServerVnodeNum();
			int minLoadServerId = 0;
			int maxLoadServerId = 0;
			for (Integer serverId : serverId2Load.keySet()) {
				if ((minLoadServerId == 0 || serverId2Load.get(serverId) < serverId2Load
						.get(minLoadServerId))
						&& !splitDestinList.contains(serverId)) {
					minLoadServerId = serverId;
				}
				if ((maxLoadServerId == 0 || serverId2Load.get(serverId) > serverId2Load
						.get(maxLoadServerId))
						&& serverId2VnodeNum.get(serverId) > 1
						&& !splitSourceList.contains(serverId)) {
					maxLoadServerId = serverId;
				}
			}
			double minLoad = 0.0;
			if (minLoadServerId != 0) {
				minLoad = serverId2Load.get(minLoadServerId);
			}
			double maxLoad = 0.0;
			if (maxLoadServerId != 0) {
				maxLoad = serverId2Load.get(maxLoadServerId);
			}
			// System.err.println("maxLoad: " + maxLoad + " minLoad: " +
			// minLoad);
			if (minLoadServerId != 0 && maxLoadServerId != 0
					&& maxLoad >= GlobalVariables.minLoadRatio * minLoad
					&& maxLoad >= GlobalVariables.minLoadToSplit) {
				splitSourceList.add(maxLoadServerId);
				splitDestinList.add(minLoadServerId);
				msg.toId = maxLoadServerId;
				msg.moveToId = minLoadServerId;
				// System.out.println("Split needed from " + msg.toId + "("
				// + serverId2VnodeNum.get(msg.toId) + ") with load "
				// + maxLoad + " to " + msg.moveToId + "("
				// + serverId2VnodeNum.get(msg.moveToId) + ") with load "
				// + minLoad);
				DhtStatistics.movementReport(access.startTime, msg.toId,
						serverId2VnodeNum.get(msg.toId), maxLoad, msg.moveToId,
						serverId2VnodeNum.get(msg.moveToId), minLoad);
			} else {
				// No need to split
				return null;
			}
			msg.msgSize = GlobalVariables.splitReqSize;
			msg.table = routeTable;
			break;
		default:
			throw new DhtException("masterStartAccess msgType unknown");
		}
		return msg;
	}

	public Vector<Message> masterProcessMsg(Message msg, Access access)
			throws DhtException {
		DhtStatistics.serverCpuReport(0,
				Math.max(idleTime, msg.getCurrentTime()) - idleTime, diskTime,
				Math.max(idleTime, msg.getCurrentTime()) + diskTime);

		if (msg.msgType == MessageType.TABLE_ACK
				|| msg.msgType == MessageType.SPLIT_FLUSH_ACK
				|| msg.msgType == MessageType.META_READ_ACK) {
			idleTime = Math.max(idleTime, msg.getCurrentTime()) + diskTime;
			return null;
		}

		Vector<Message> msgList = new Vector<Message>();
		Message newMsg = new Message();

		newMsg.accessId = msg.accessId;
		newMsg.fromId = msg.toId;
		newMsg.toId = msg.fromId;
		newMsg.generateTime = msg.getCurrentTime();

		idleTime = Math.max(idleTime, newMsg.generateTime);
		newMsg.waitingCost = idleTime - newMsg.generateTime;
		newMsg.processCost = diskTime;
		newMsg.sendTime = newMsg.generateTime + newMsg.waitingCost
				+ newMsg.processCost;
		idleTime = newMsg.sendTime;

		switch (msg.msgType) {
		case META_READ_REQ:
			newMsg.msgSize = GlobalVariables.metaReadRespSize;
			newMsg.table = routeTable;
			newMsg.msgType = MessageType.META_READ_RESP;
			break;
		case TABLE_REQ:
			updateStatistics(msg.fromId, msg.generateTime, msg.table);
			// if (routeTable.getServerLoads().get(msg.fromId) >=
			// GlobalVariables.minLoadToSplit) {
			// System.err.println(msg.fromId + " " + msg.getCurrentTime()
			// + " " + routeTable.getServerLoads().get(msg.fromId));
			// }
			newMsg.msgSize = GlobalVariables.tableRespSize;
			newMsg.table = routeTable;
			newMsg.msgType = MessageType.TABLE_RESP;
			DhtStatistics.serverReport(0, 0, 0, 1, 0);
			break;
		case SPLIT_RESP:
			newMsg.msgSize = GlobalVariables.splitAckSize;
			newMsg.msgType = MessageType.SPLIT_ACK;
			break;
		case SPLIT_FLUSH_REQ:
			cloneRouteTable();
			routeTable.increaseVersion(msg.fromId);
			routeTable.increaseVersion(msg.moveSourceId);
			splitSourceList.remove(msg.moveSourceId);
			splitSourceList.remove(msg.fromId);
			for (VNodeV2 vnode : msg.vecVnodes) {
				routeTable.vnodeId2Vnode.put(vnode.id, (VNodeV2) vnode.clone());
			}
			newMsg.msgSize = GlobalVariables.splitFlushRespSize;
			newMsg.msgType = MessageType.SPLIT_FLUSH_RESP;
			DhtStatistics.serverReport(0, 0, 0, 0, 1);
			break;
		case META_READ_ACK:
			return null;
		case TABLE_ACK:
			return null;
		case SPLIT_FLUSH_ACK:
			return null;
		default:
			throw new DhtException("masterProcessMsg msgType unknown");
		}
		msgList.add(newMsg);
		return msgList;
	}

	private void cloneRouteTable() {
		this.routeTable = (RouteTableV2) this.routeTable.clone();
	}

	private void updateStatistics(int serverId, double lastReportTime,
			RouteTableV2 serverTable) throws DhtException {
		routeTable.updateVNodeStatistics(serverId, lastReportTime,
				serverTable.vnodeId2Vnode);
	}

}
